package cn.admin.configcenter.registry.service.client;

import cn.admin.config.common.Revision;
import cn.admin.config.common.protocol.*;
import cn.admin.configcenter.registry.IOCFactory;
import cn.admin.configcenter.registry.common.constants.WarningCode;
import cn.admin.configcenter.registry.common.info.ClientConnection;
import cn.admin.configcenter.registry.common.info.GroupId;
import cn.admin.configcenter.registry.common.util.ConfigUtil;
import cn.admin.configcenter.registry.service.DataService;
import cn.admin.remoting.Connection;
import cn.admin.remoting.RequestProcessor;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ClientRequestJob implements Runnable {

    private static final Logger log = LoggerFactory.getLogger(ClientRequestJob.class);
    private static final DataService dataService = IOCFactory.getDataService();
    final ProtocolPackage pp;
    final Connection conn;
    final RequestProcessor.AppResponseOutput output;
    final long createTime;

    public ClientRequestJob(ProtocolPackage _pp, RequestProcessor.AppResponseOutput _output) {
        this.pp = _pp;
        this.output = _output;
        this.conn = _output.getConnection();
        this.createTime = System.currentTimeMillis();
        if (this.pp.countElements() >= 50) {
            log.warn("too many ({}) elements from {}", this.pp.countElements(),
                    ConfigUtil.generateHostId(this.output.getConnection()));
        }

    }

    @Override
    public void run() {
        long processBegin = System.currentTimeMillis();
        long pendingTime = processBegin - this.createTime;
        if (pendingTime >= 600L) {
            log.warn("[drop-client-request] Pending too long in queue. " + pendingTime);
        } else {
            ProtocolPackage resp = new ProtocolPackage();
            ClientConnection client = ClientConnectionWorker.getWorker().addNativeClient(this.conn);
            for (ProtocolElement element : this.pp) {
                if (element instanceof PublisherRegReqPacket) {
                    this.process((PublisherRegReqPacket) element,client,resp);
                } else if (element instanceof PublisherDataElement) {
                    this.process((PublisherDataElement)element,client,resp);
                } else if (element instanceof SubscriberRegReqPacket) {
                    this.process((SubscriberRegReqPacket)element,client,resp);
                } else if (element instanceof AttributeElement) {
                    this.process((AttributeElement)element,client);
                }
            }
        }
    }

    private void process(AttributeElement attributeElement,ClientConnection client) {
        if ("!ZIP".equals(attributeElement.name)) {
            client.setSupportGzip();
        }
    }

    private void process(PublisherDataElement dataElement,ClientConnection client,
                         ProtocolPackage resp) {
        String dataId = dataElement.dataId;
        if (!this.isHsfVersionRequest(dataId)) {
            Revision revision = dataElement.getRevision();
            List<Swizzle> data = dataElement.getShelteredData();
            dataService.publishData(client,dataId,dataElement.clientId,data,revision);
        }

        resp.addElement(new UserDataResultElement(dataElement));
    }

    private void process(SubscriberRegReqPacket packet,ClientConnection client,
                         ProtocolPackage resp) {

        SubscriberRegElement sre = null;
        for (ProtocolElement element : packet) {
            if (element instanceof SubscriberRegElement) {
                sre = (SubscriberRegElement) element;
                break;
            }
        }

        if (sre == null) {
            log.error("package error: SubscriberRegElement not found. {}" + client);
        } else {
            String dataId = sre.dataId;
            if (!this.isHsfVersionRequest(dataId)) {
                Map<String,Object> attributeMap = this.getAttributeMap(packet);
                dataService.addSubscriber(client,
                        this.getGroupId(sre.dataId,attributeMap),sre.clientId);
            }

            resp.addElement(new SubscriberRegResultElement(dataId,sre.clientId));
        }

    }

    private void process(PublisherRegReqPacket packet,ClientConnection client,
                         ProtocolPackage resp) {
        PublisherRegElement pre = null;
        for (ProtocolElement element : packet) {
            if (element instanceof PublisherRegElement) {
                pre = (PublisherRegElement) element;
                break;
            }
        }

        if (pre == null) {
            log.error("package error: PublisherRegElement not found. {}" + client);
        } else {
            Map<String,Object> attributeMap = this.getAttributeMap(packet);
            GroupId group = this.getGroupId(pre.dataId,attributeMap);
            boolean declarePersist = this.isPersistentData(attributeMap);

            if (isNotifySubinfo(group)) {
                log.info("[notify-subinfo] {} add publisher {} at datum[{}] at clientId[{}]",
                        client, group, pre.datumId, pre.clientId);
            }

            if (!this.isHsfVersionRequest(pre.dataId)) {
                if (group.isPersist() == declarePersist) {
                    dataService.addPublisher(client,group,pre.clientId,pre.datumId);
                }else {
                    WarningCode.PERSISTENCY_CONFLICT.warning("[persistency-conflict] "
                            + pre.dataId + " is " + group.getPersistString() +
                            " according to white list, but " + client + " declares otherwise.");
                }
            }

            resp.addElement(new PublisherRegResultElement(pre.dataId,pre.clientId));
        }
    }

    private boolean isHsfVersionRequest(String dataId) {
        return dataId.startsWith("hsf.") && dataId.endsWith(".version");
    }

    private boolean isPersistentData(Map<String,Object> attributeMap) {

        Object val = attributeMap.get(AttributeElement.ATTRIBUTE_PERSISTENCY);
        return null != val ? (Boolean)val : false;

    }

    private GroupId getGroupId(String dataId,Map<String,Object> attributeMap) {

        String groupId = (String) attributeMap.get(AttributeElement.ATTRIBUTE_GROUP);
        if (StringUtils.isEmpty(groupId)) {
            groupId = "DEFAULT_GROUP";
        }

        return GroupId.getSingleton(dataId,groupId);

    }

    private Map<String,Object> getAttributeMap(DataClientRegReqPacket packet) {

        Map<String,Object> result = new HashMap<>();
        for (ProtocolElement element : packet) {
            if (element instanceof AttributeElement) {
                AttributeElement attr = (AttributeElement)element;
                String name = attr.getName();
                if (StringUtils.isNotBlank(name)) {
                    result.put(name, attr.getValue());
                }
            }
        }

        return result;

    }

    static boolean isNotifySubinfo(GroupId group) {
        return group.dataId.startsWith("NS_Notify_SubscriptionInfo_Binding");
    }
}
